package org.example.java_kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

//消息发送方
public class MyProducer {
    //主题名称
    private final static String TOPIC_NAME = "partition10";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1. 设置参数
        Properties props = new Properties();
        //指定服务器配置【ip：端口】
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.2.135:9092,192.168.2.135:9093,192.168.2.135:9094");

        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        //把发送消息的value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        //ack
        props.put(ProducerConfig.ACKS_CONFIG,"1");
        //失败重试 默认间隔100ms，保证消息可靠性，也可能造成消息重复，比如网络抖动，所以需要在接受者处做好幂等性处理
        //重试次数 3次
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        //重试时间
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,300);
        //缓冲区
        //设置发送消息的本地缓冲区，设置之后消息会先发送到本地缓冲区，可以提高消息发送性能，默认值33554432，即32M
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
        //本地线程会从缓冲区取数据，批量发送到broker
        //设置批量发送消息的大小，默认16k，即16384，也就是说一个batch满16k就发出去
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        //LINGER_MS_CONFIG
        //默认值0，意思是立即发送，但是影响性能
        //一般设置10ms左右，就是说消息发送完成后会进入一个本地batch，10ms内这个batch满了16k就随batch一同发出
        //10ms内没满16k也发出
        props.put(ProducerConfig.LINGER_MS_CONFIG,10);
        //2. 创建生产消息的客户端，传入参数
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i=0;i<5;i++) {
            //3. 创建消息；key：作用是决定了往哪个分区上发，value：具体要发送的消息内容
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                    TOPIC_NAME,1, "myKeyValue1", "he00lloKafka");
            System.out.println(1);
            //4. 同步发送消息，得到消息发送的元数据并输出
            RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());

            //异步发送
            Future future = producer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e != null) {
                        System.out.println(e);
                    }
                    if (recordMetadata != null) {
                        System.out.println("异步方式发送消息结果：" + "topic-" + recordMetadata.topic() + "|partition-"
                                + recordMetadata.partition() + "|offset-" + recordMetadata.offset());
                    }
                }
            });
        }

//        Object o = future.get();
//        Thread.sleep(100);
        producer.close();
    }
}

